#include "Consumer.h"

#include <unordered_map>
#include <iostream>

#include "FileStream.h"
namespace streamdb {


struct Subject  : public IIterator<std::string> {
    Subject(const std::string & path, int64_t offset)
    : offset(offset)
    , stream(std::make_shared<FileStream>(path))
    {
    }
    ~Subject(){}


    bool hasNext()const override{
        return cur->hasNext();
    }
    std::string * next() override{
        return &(cur->next());
    }

    Status open(){
        auto s = stream->open();
        if(!s.isOk()){
            std::cerr << "open failed, stream:" << stream->getName() << std::endl;
            return s;
        }
        int real = (offset > (stream->getSize()-1)) ? (stream->getSize()-1) : offset;
        cur = stream->advance(real); //assert(offset <= stream.size()) TODO: offset > stream.size?????
        assert(cur != nullptr);
        return Status::OK();
    }
    void close(){
        return stream->close();
    }

    const int64_t offset;
    std::shared_ptr<FileStream> stream;
    std::shared_ptr<SequenceStream<std::string>::IteratorType> cur;
    
};

class ConsumerImp : public Consumer {
public:
    typedef std::unordered_map<std::string, std::shared_ptr<Subject>> SubjectMap;

    class iterator : public IIterator<std::string> {
    public:

        explicit iterator(SubjectMap & subjects) 
        : subjects(&subjects) 
        , iter(subjects.begin())
        {
        }
        ~iterator() override{
        }

        iterator(iterator && other) {
            subjects = other.subjects;
            other.subjects = nullptr;
        }
        iterator(const iterator &other) = delete;
        void operator=(const iterator & other) = delete;

        bool hasNext() const override{
            return iter != subjects->end();
        }
        std::string * next() override{
            while(iter != subjects->end()){
                if(iter->second->hasNext()){
                    return iter->second->next();
                }
                else{
                    ++iter;
                }
            }
            return nullptr;
        }
    private:
       SubjectMap * subjects;
       SubjectMap::iterator iter;
    };


    ConsumerImp(const std::string & dir) 
    : dir(dir)
    , isOpen(false)
    {
        open();
    }
    ~ConsumerImp() override {
        close();
    }


    const std::string &getDirectory() const override{
        return dir;
    }

    Status open() override{ 
        if(!isOpen){ 
            isOpen = true;
        }
        return Status::OK();
    }

    Status subscribe(const std::string & topic, int64_t offset) override{
        if(!isOpen){
            std::cerr << "ERROR: not opened" << std::endl;
            return Status::InvalidArgument("not opened");
        }

        auto iter = subjects.find(topic);
        if(iter == subjects.end()){
            auto sub = std::make_shared<Subject>(dir + "/" + topic, offset);
            auto s = sub->open();
            if(!s.isOk()){
                std::cerr << "sub open failed" << std::endl;
                return Status::InvalidArgument("sub open failed");
            }
            subjects.insert(std::make_pair(topic, sub));
        }
        return Status::OK();
    }

    std::unique_ptr<IIterator<std::string>> all() override {
        if(!isOpen){
            std::cerr << "ERROR: not opened" << std::endl;
            return std::move(std::unique_ptr<IIterator<std::string>>());
        }
        return std::move(std::unique_ptr<IIterator<std::string>>(new iterator(subjects)));
    }

    std::unique_ptr<IIterator<std::string>> poll(std::chrono::milliseconds timeout) override{
        if(!isOpen){
            std::cerr << "ERROR: not opened" << std::endl;
            return std::move(std::unique_ptr<IIterator<std::string>>());
        }
        //condition.timeWait(timeout);
        return std::move(std::unique_ptr<IIterator<std::string>>(new iterator(subjects)));
    }
    void close() override{
        if(isOpen){
            for(auto & iter : subjects){
                iter.second->close();
            }
            isOpen = false;
        }
    }

private:
    std::string dir;
    bool isOpen;
    std::unordered_map<std::string, std::shared_ptr<Subject>> subjects;
};

std::shared_ptr<Consumer> makeConsumer(const std::string & dir) {
    return std::make_shared<ConsumerImp>(dir);
}

} //end of namespace streamdb
